package edu.xhu.homepage.bloomFilter;

import com.google.common.hash.Funnel;
import com.google.common.hash.Hashing;
import com.google.common.primitives.Longs;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.*;

//线程安全之布隆
@Slf4j
public class BillBloomFilterV2<T> {
    //redis中bitmap的key
    private final String bitmapKey;

    private static final long bitmapSize = 20_000; // 10万位,太大,长度是16000,目前只支持2000x8=16000个视频

    //hash函数的个数
    private final int numHashFunctions;

    //待查询关键字缓存，预期结果：{ thread1=[k1,k2,k3], thread2=[k1,k2] }
    private final Map<String, List<T>> keysToQueryCacheMap;

    //待插入关键字缓存，预期结果：{ thread1=[k1,k2,k3], thread2=[k1,k2] }
    private final Map<String, List<T>> keysToInsertCacheMap;

    // Google Guava 库中的一个接口，用于将对象转换为字节流，以便与布隆过滤器（Bloom Filter）一起使用。
    private final Funnel<T> funnel;

    private final StringRedisTemplate redisTemplate;

    public BillBloomFilterV2(String bitmapKey, int numHashFunctions, Funnel<T> funnel, StringRedisTemplate redisTemplate) {
        //首先判断redis中是否存在bitmapKey，如果不存在抛异常
        if (!redisTemplate.hasKey(bitmapKey))
            throw new RuntimeException("布隆过滤器异常，redis中不存在bitmapKey: " + bitmapKey);

        this.bitmapKey = bitmapKey;
        this.numHashFunctions = numHashFunctions;
        this.funnel = funnel;
        this.redisTemplate = redisTemplate;

        this.keysToQueryCacheMap = new HashMap<>();
        this.keysToInsertCacheMap = new HashMap<>();
    }

    //添加查询关键字至缓存keysToQueryCacheMap.get(threadId)
    public void addKeysToQueryCache(T key) {
        // 获取当前线程的 ID
        String threadId = String.valueOf(Thread.currentThread().getId());

        //keysToQueryCacheMap中不存在当前线程的querykey缓存
        if (!keysToQueryCacheMap.containsKey(threadId))
            keysToQueryCacheMap.put(threadId, new ArrayList<>());

        keysToQueryCacheMap.get(threadId).add(key);
    }


    /**
     * 利用redis管道查询 keysToQueryCacheMap.get(threadId) 中缓存的querykey是否存在于布隆过滤器，
     * 查询完毕后清空 keysToQueryCacheMap.get(threadId) 缓存的querykey
     *
     * @return map={ "true"=[k1,k2,k3], "false"=[k4,k5] }，
     * [k1,k2,k3]存在于布隆过滤器，视频被推荐过
     * [k4,k5]   不存在于布隆过滤器，视频未被推荐过
     */
    public Map<String, List<T>> checkKeysInBloomFilterWithPipeline() {

        //首先判断redis中是否存在bitmapKey，如果不存在抛异常
        if (!redisTemplate.hasKey(bitmapKey))
            throw new RuntimeException("布隆过滤器异常，redis中不存在bitmapKey: " + bitmapKey);

        //存储查询结果
        Map<String, List<T>> queryResultsMap = new HashMap<>();
        queryResultsMap.put("true", new ArrayList<>());
        queryResultsMap.put("false", new ArrayList<>());

        // 获取当前线程的 ID
        String threadId = String.valueOf(Thread.currentThread().getId());

        //当前线程未缓存querykey，直接返回
        if (!keysToQueryCacheMap.containsKey(threadId) || keysToQueryCacheMap.get(threadId).size() == 0)
            return queryResultsMap;

        //开始查询，tempQueryResults为中间查询结果
        //tempQueryResults中间结果示例(numHashFunctions个为一组)：
        //[false, false, false, false, false, false, false, false, false, false, false, false, false, false, false]
        List<Object> tempQueryResults = redisTemplate.executePipelined((RedisCallback<Boolean>) connection -> {
            for (T key : keysToQueryCacheMap.get(threadId)) {
                //将key转换为字节流
                byte[] bytes = Hashing.murmur3_128().hashObject(key, funnel).asBytes();

                //增量式构造 key 的 numHashFunctions 个哈希值。
                long initialHash = this.lowerEight(bytes); // 初始哈希值
                long hashIncrement = this.upperEight(bytes);  // 每次增加的哈希值
                long cumulativeHash = initialHash; // 累计的哈希值

                //开始查询
                for (int i = 0; i < numHashFunctions; i++) {
                    cumulativeHash += hashIncrement;
                    connection.getBit(bitmapKey.getBytes(), (cumulativeHash & Long.MAX_VALUE) % bitmapSize);
                }
            }
            return null;
        });


        //开始根据中间查询结果打包最终结果
        int counter = 0;
        int keyIndex = 0; // key在keysToQueryCacheMap.get(threadId)中的索引
        boolean isExistInBloom = true; //标识 key 是否存在于布隆过滤器
        //开始处理中间查询结果，(数组中numHashFunctions个元素为一组)
        for (Object o : tempQueryResults) {

            //出现一个false，说明 key 不在布隆过滤器中
            if (!Boolean.parseBoolean(o.toString()))
                isExistInBloom = false;

            counter += 1;

            //遍历完了一组中间结果，即遍历完了一个key的所有哈希值
            if (counter % numHashFunctions == 0) {
                if (!isExistInBloom)
                    //缓存的keysToQueryCacheMap.get(threadId)中第 keyIndex 个 key 不存在于布隆过滤器中,(视频未被推荐过)
                    queryResultsMap.get("false").add(keysToQueryCacheMap.get(threadId).get(keyIndex));
                else
                    //缓存的keysToQueryCacheMap.get(threadId)中第 keyIndex 个 key 存在于布隆过滤器中,(视频被推荐过)
                    queryResultsMap.get("true").add(keysToQueryCacheMap.get(threadId).get(keyIndex));

                //刷新标志变量
                isExistInBloom = true;

                //刷新key索引
                keyIndex += 1;
            }
        }

        //查询完毕后，清空keysToQueryCacheMap.get(threadId)中缓存的key
        keysToQueryCacheMap.remove(threadId);

        return queryResultsMap;
    }


    //添加插入关键字至缓存keysToInsertCacheMap.get(threadId)
    public void addKeysToInsertCache(T key) {
        // 获取当前线程的 ID
        String threadId = String.valueOf(Thread.currentThread().getId());

        //keysToInsertCacheMap中不存在当前线程的insertkey缓存
        if (!keysToInsertCacheMap.containsKey(threadId))
            keysToInsertCacheMap.put(threadId, new ArrayList<>());

        keysToInsertCacheMap.get(threadId).add(key);
    }


    /**
     * 利用redis管道将keysToInsertCacheMap.get(threadId)中缓存的insertkey插入至布隆过滤器，
     * 插入完毕后清空keysToInsertCacheMap.get(threadId)
     */
    public void insertKeysToBloomFilterWithPipeline() {
        // 获取当前线程的 ID
        String threadId = String.valueOf(Thread.currentThread().getId());

        //当前线程未缓存insertkey，直接返回
        if (!keysToInsertCacheMap.containsKey(threadId) || keysToInsertCacheMap.get(threadId).size() == 0)
            return;

        redisTemplate.executePipelined((RedisCallback<Boolean>) connection -> {

            for (T key : keysToInsertCacheMap.get(threadId)) {
                //将key转换为字节流
                byte[] bytes = Hashing.murmur3_128().hashObject(key, funnel).asBytes();

                //增量式构造 key 的 numHashFunctions 个哈希值。
                long initialHash = this.lowerEight(bytes); // 初始哈希值
                long hashIncrement = this.upperEight(bytes);  // 每次增加的哈希值
                long cumulativeHash = initialHash; // 累计的哈希值

                //开始插入key至布隆过滤器
                for (int i = 0; i < numHashFunctions; i++) {
                    cumulativeHash += hashIncrement;
                    long bitIndex = (cumulativeHash & Long.MAX_VALUE) % bitmapSize;
                    connection.setBit(bitmapKey.getBytes(), bitIndex, true);
                }
            }
            return null;
        });

        //清空keysToInsertCacheMap.get(threadId)中缓存的key
        keysToInsertCacheMap.remove(threadId);
    }

    //清空当前线程的 queryKeys 缓存
    public void clearCurrentThreadQueryKeys() {
        // 获取当前线程的 ID
        String threadId = String.valueOf(Thread.currentThread().getId());

        keysToQueryCacheMap.remove(threadId);
    }


    //清空当前线程的 insertKeys 缓存
    public void clearCurrentThreadInsertKeys() {
        // 获取当前线程的 ID
        String threadId = String.valueOf(Thread.currentThread().getId());

        keysToInsertCacheMap.remove(threadId);
    }


    //从字节数组的前 8 个字节（索引 0 到 7）构建一个 long 值
    private long lowerEight(byte[] bytes) {
        return Longs.fromBytes(bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]);
    }

    //从字节数组的后 8 个字节（索引 8 到 15）构建一个 long 值。
    private long upperEight(byte[] bytes) {
        return Longs.fromBytes(bytes[15], bytes[14], bytes[13], bytes[12], bytes[11], bytes[10], bytes[9], bytes[8]);
    }
}
